package com.shujia.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    /**
      * 创建spark 环境
      *
      */

    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("wc")


    val sc = new SparkContext(conf)


    /**
      * 创建spark streaming环境
      * 指定每隔多久计算一次
      *
      */

    val ssc = new StreamingContext(sc, Durations.seconds(5))


    /**
      * 读取数据
      * 读取socket  数据
      * nc -lk  8888  用于测试
      *
      */

    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)


    /**
      * 处理数据
      *
      */

    val countDS: DStream[(String, Int)] = linesDS
      .flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey(_ + _)


    countDS.print()


    /**
      * 启动spark streaming
      *
      */

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
